//package com.pubinfo.passbook.common.utils;
//
//import com.pubinfo.passbook.common.config.KafkaConfig;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.apache.kafka.clients.producer.*;
//import org.springframework.context.annotation.Bean;
//
//import javax.annotation.Resource;
//import java.time.Duration;
//import java.util.Arrays;
//import java.util.Properties;
//
//public class KafkaUtil {
//  @Resource
//  KafkaConfig config;
//  public void send(String key){
//    Producer<String, String> producer = new KafkaProducer<String, String>(kafkaProducer());
//    int i = 0;
//    boolean stopFlag = false;
//    while (!stopFlag) {
//      String message = "hello " + (i++);
//      //如指定key，将会进根key hash到指定分区,没特别需求，建议key为null，由kafka自动分配分区，以免造成各个分区消息分布不均衡
//      //构造Producer对象，注意，该对象是线程安全的，一般情况下，一个进程内可共用一个Producer对象
//      ProducerRecord record = new ProducerRecord<>(config.getTopic(), key, message);
//
//      //建议采用callback的方式，获取发送结果
//      producer.send(record, new Callback() {
//        @Override
//        public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
//          if (exception != null) {
//            //todo log and retry
//            System.out.printf("send  exception. %s%n", exception.getMessage());
//            return;
//          }
//          //todo business process...
//          System.out.printf("send message success. topic=%s,key=%s,offset=%s,partition=%d %n",
//              recordMetadata.topic(), key, recordMetadata.offset(), recordMetadata.partition());
//        }
//      });
//    }
//    producer.close();//close
//  }
//
//
//  public Properties kafkaProducer() {
//    Properties properties = new Properties();
//    properties.put("security.protocol", "SASL_PLAINTEXT");
//    properties.put("sasl.mechanism", "SCRAM-SHA-512");
//    properties.put("bootstrap.servers", config.getBrokers() );
//    properties.put("acks", "all");// -1：all ack;1：leader ack;0：none ack
//    properties.put("retries", 3);//失败重试次数
//    properties.put("batch.size", 262144);
//    properties.put("linger.ms", 10);
//    properties.put("buffer.memory", 67108864);
//    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//    properties.put("sasl.jaas.config", config.getJaas());
//    //metric.reporters用于上报客户端指标数据，可选（不设置则无法在管理台查询客户端指标信息），建议设置;
//    //如未引入ctg-kafka-client-metrics-reporter依赖，以下代码将导致报错
//    properties.put("metric.reporters", "com.ctg.kafka.clients.reporter.KafkaClientMetricsReporter");
//    //每个客户端实例使用不同的client.id(建议使用与应用、业务相关的名字作为前缀)
//    properties.put("client.id", "producer-demo_" + System.currentTimeMillis());
//    return properties;
//  }
//
//}
